package com.ehotting.mqbd.core.mq.consumer.action;

import cn.hutool.core.lang.Assert;
import com.ehotting.eaf.core.snow.SnowflakeUtil;
import com.ehotting.mqbd.api.bean.MQDataAble;
import com.ehotting.mqbd.api.bean.MqApplicationContext;
import com.ehotting.mqbd.api.enums.DirectionEnum;
import com.ehotting.mqbd.api.enums.TopicEnum;
import com.ehotting.mqbd.core.call.CallerAble;
import com.ehotting.mqbd.core.logger.LOGGER_KEY;
import com.ehotting.mqbd.core.logger.MqMsgLogger;
import com.ehotting.mqbd.core.mq.listener.MQListenerConcurrently;
import com.ehotting.mqbd.core.mq.listener.MQListenerOrderly;
import com.ehotting.mqbd.core.manager.ModuleManager;
import com.ehotting.mqbd.core.manager.SolverAllocator;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.MDC;


public  class ConsumerRequestAction extends ConsumerActionAbstract {

    private CallerAble caller;

    private MqApplicationContext mqApplicationContext;

    private TopicEnum topic;

    private DirectionEnum direction;

    private String solverIp;

    private SolverAllocator allocator;

    public ConsumerRequestAction(SolverAllocator allocator,
                                 CallerAble caller,
                                 MqApplicationContext mqApplicationContext,
                                 TopicEnum topic,
                                 DirectionEnum direction, String solverIp, ModuleManager moduleManager, MqMsgLogger mqMsgLogger){
        this.allocator = allocator;
        this.caller = caller;
        this.mqApplicationContext = mqApplicationContext;
        this.topic = topic;
        this.direction = direction;
        this.solverIp = solverIp;
        this.moduleManager = moduleManager;
        this.mqMsgLogger = mqMsgLogger;
    }



    @Override
    public void invoke(MessageExt msg) throws Exception {
        logger.info("Start invoke.");
        MDC.put(LOGGER_KEY.TRACE_ID, SnowflakeUtil.nextId64());
        MDC.put(LOGGER_KEY.MODULE,mqApplicationContext.getModule());

        Assert.notNull(msg);
        Assert.notNull(caller);
        Assert.notNull(topic);
        Assert.notNull(mqApplicationContext);

        String str =  new String(msg.getBody());
        MQDataAble data = convert(str,msg.getMsgId());
        logger.info("Start allocator invoke.");

        Assert.notNull(data.getRequest(),"data["+data.getModule()+"] request object is null.");

        allocator.invoke(caller, data);

        logger.info("Finish allocator invoke.");

        extInvoke(data);

        logger.info("Finish invoke.");
    }



    public String getHost(){
        return mqApplicationContext.getLocalHost();
    }

    @Override
    public String getModule() {
        return mqApplicationContext.getModule();
    }

    @Override
    public DirectionEnum getDirection() {
        return direction;
    }

    public TopicEnum getTopic() {
        return topic;
    }

    @Override
    public String getSolverIp() {
        return solverIp;
    }

    @Override
    public void extInvoke(MQDataAble data) {

    }

    @Override
    public MessageListener getMessageListener() {
        return this.getTopic()== TopicEnum.BLOCK?
                new MQListenerOrderly(this)
                :
                new MQListenerConcurrently(this);
    }

    @Override
    public MqApplicationContext getMqApplicationContext() {
        return mqApplicationContext;
    }
}
